/*
 * CloudClient.cpp
 *
 *  Created on: Jan 23, 2012
 *      Author: hduser
 */

#include<iostream>
#include<fstream>
#include<string>
#include "hdfs.h"
#include "Connection.h"
#include "CloudClient.h"

using namespace std;
using namespace utilities;
using namespace logging;

namespace cloud {

CloudClient::CloudClient(Connection* conn) {
	cloudConn = conn;
	init();
}

CloudClient::CloudClient(char* hostname, int port_number) {
	cloudConn = new Connection(hostname, port_number);
	cloudConn->Open();
	init();
}

CloudClient::~CloudClient() {
	if (cloudConn != NULL) {
		cloudConn->Close();
	}
}

/*********************************************************/
/* PUBLIC FUNCTIONS           							 */
/*********************************************************/

void CloudClient::init() {
	logger = new Logger();
}

int CloudClient::writeFileToHDFS(char* localfileName, char* hdfsfileName) {

	try {

		if (cloudConn == NULL)
			return -1;

		if (cloudConn->cloudFS == NULL)
			return -2;

		size_t num_read_bytes;
		char* localbuffer;
		streampos currentpos = 0;

		ifstream fl(localfileName, ios::in | ios::binary);
		fl.seekg(0, ios::end );
		size_t filelength = fl.tellg();

		hdfsFile writeFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_WRONLY|O_CREAT, 0, 0, 0);

		if(!writeFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for writing!"<<endl;
			return -1;
		}

		FileUtilities* fu = new FileUtilities();

		while (currentpos < filelength) {

			//cout<<"currentpos: "<<currentpos<<endl;
			//cout<<"filelength: "<<filelength<<endl;

			if (filelength - currentpos < BLOCK_SIZE)
				num_read_bytes = filelength - currentpos;
			else
				num_read_bytes = BLOCK_SIZE;

			num_read_bytes = fu->readBytes(&fl, &localbuffer, currentpos, num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsFlush(cloudConn->cloudFS, writeFile);
				hdfsCloseFile(cloudConn->cloudFS, writeFile);
				hdfsDelete(cloudConn->cloudFS, hdfsfileName);
				fl.close();
				return -1;
			}

			size_t num_written_bytes = hdfsWrite(cloudConn->cloudFS, writeFile, (void*)localbuffer, num_read_bytes);

			currentpos = currentpos + num_read_bytes;
		}

		if (hdfsFlush(cloudConn->cloudFS, writeFile)) {
			cerr<<"Failed to 'flush' "<<hdfsfileName<<endl;
			return -1;
		}
		hdfsCloseFile(cloudConn->cloudFS, writeFile);
		fl.close();
		return 0;
	} catch(exception ex) {
		return -1;
	}
}

int CloudClient::getFileFromHDFS(char*  hdfsfileName, char* localfilename) {

	try {

		if (cloudConn == NULL)
					return -1;

		if (cloudConn->cloudFS == NULL)
			return -2;

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);
		char* buffer;
		size_t num_read_bytes;

		if (exists)	{
			cerr<<"Failed to validate existence of: "<<hdfsfileName<<endl;
			exit(-1);
		}

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"Failed to determine size of: "<<hdfsfileName<<endl;
			exit(-1);
		}

		hdfsFile readFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_RDONLY, 0, 0, 0);
		if (!readFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for reading!"<<endl;
			exit(-1);
		}

		size_t seekPos = 0;
		size_t currentPos = 0;

		seekPos = currentPos;
		if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
			cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
			exit(-1);
		}

		if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
			cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
			exit(-1);
		}

		ofstream fl(localfilename, ios::out | ios::binary);
		FileUtilities* fu = new FileUtilities();

		while (currentPos < hdfsfilesize) {

			//cout<<"currentPos: "<<currentPos<<endl;
			//cout<<"hdfsfilesize: "<<hdfsfilesize<<endl;

			if (hdfsfilesize - currentPos < BLOCK_SIZE)
				num_read_bytes = hdfsfilesize - currentPos;
			else
				num_read_bytes = BLOCK_SIZE;

			buffer = new char[num_read_bytes];
			num_read_bytes = hdfsPread(cloudConn->cloudFS, readFile, currentPos, (void*)buffer,num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsCloseFile(cloudConn->cloudFS, readFile);
				fl.close();
				return -1;
			}

			fu->writeBytes(&fl, buffer, num_read_bytes);

			currentPos = currentPos + num_read_bytes;

			/*
			seekPos = currentPos;
			if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
				cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
				exit(-1);
			}

			if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
				cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
				exit(-1);
			}
			*/
		}

		hdfsCloseFile(cloudConn->cloudFS, readFile);
		fl.close();
		return 0;

	} catch(exception ex) {
		return -1;
	}
}

int CloudClient::getChunkFromHDFS(char*  hdfsfileName, int chunkIndex, int totalChunks, char** returnBuffer) {

	try {

		if (cloudConn == NULL)
			return -2; // connection error

		if (cloudConn->cloudFS == NULL)
			return -2; // connection error

		if (chunkIndex > totalChunks ) {
			cerr<<"chunk Index should be less than total chunks: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "chunk Index should be less than total chunks", hdfsfileName);
			return -11; // custom error
		}

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);

		if (exists)	{
			cerr<<"file not found in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "file not found in cloud", hdfsfileName);
			return -3; // file not exists
		}

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"cannot deterine size of the file in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "cannot deterine size of the file in cloud", hdfsfileName);
			return -3; // cannot determine file size
		}

		char* buffer;
		size_t num_read_bytes;

		hdfsFile readFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_RDONLY, 0, 0, 0);
		if (!readFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for reading!"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "cannot open file in cloud", hdfsfileName);
			return -3; // cannot open file
		}

		size_t seekPos = 0;
		size_t currentPos = chunkIndex * CHUNK_SIZE;

		seekPos = currentPos;
		if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
			cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "Failed to seek file in cloud", hdfsfileName);
			return -3; // seek error
		}

		if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
			cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "failed to get current file position correctly", hdfsfileName);
			return -3; // seek error
		}

		//ofstream fl(localfilename, ios::out | ios::binary);
		//FileUtilities* fu = new FileUtilities();

		size_t maxCount = (((chunkIndex + 1) * CHUNK_SIZE) < hdfsfilesize) ? ((chunkIndex + 1) * CHUNK_SIZE) : hdfsfilesize;

		//char* returnBuffer; // = (char*)malloc(BLOCK_SIZE);
		size_t returnBufferSize = 0;

		while (currentPos < maxCount) {

			//cout<<"currentPos: "<<currentPos<<endl;
			//cout<<"hdfsfilesize: "<<hdfsfilesize<<endl;

			if (maxCount - currentPos < BLOCK_SIZE)
				num_read_bytes = maxCount - currentPos;
			else
				num_read_bytes = BLOCK_SIZE;

			buffer = new char[num_read_bytes];
			num_read_bytes = hdfsPread(cloudConn->cloudFS, readFile, currentPos, (void*)buffer,num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsCloseFile(cloudConn->cloudFS, readFile);
				//fl.close();
				logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "read failed", hdfsfileName);
				return -3; // file error
			}

			size_t tempPos = returnBufferSize;
			returnBufferSize += num_read_bytes;
			*returnBuffer = (char*)realloc((void*)*returnBuffer, returnBufferSize);
			memcpy(*returnBuffer + tempPos, buffer, num_read_bytes);

			//fu->writeBytes(&fl, buffer, num_read_bytes);

			currentPos = currentPos + num_read_bytes;
		}

		hdfsCloseFile(cloudConn->cloudFS, readFile);
		//fl.close();
		return 0;

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "CloudClient::calculateNumberOfChunks", "Caught exception");
		return -1;
	}
}

int CloudClient::calculateNumberOfChunks(char*  hdfsfileName) {

	try {

		if (cloudConn == NULL)
			return -1;

		if (cloudConn->cloudFS == NULL)
			return -2;

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);

		if (exists)	{
			cerr<<"file not found in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::calculateNumberOfChunks", "file not found in cloud", hdfsfileName);
			return -3; // file not exists
		}

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"cannot deterine size of the file in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::calculateNumberOfChunks", "cannot deterine size of the file in cloud", hdfsfileName);
			return -4; // cannot determine file size
		}

		return (int)hdfsfilesize / CHUNK_SIZE;

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "CloudClient::calculateNumberOfChunks", "Caught exception");
		return -1;
	}
}

}
